package com.xiguaji.tiktok.task.blogger.event;

import com.google.common.eventbus.AllowConcurrentEvents;
import com.google.common.eventbus.Subscribe;
import com.gugee.tiktok.common.model.AwemeScore;
import com.gugee.tiktok.common.model.dal.blogger.BloggerMainRepository;
import com.gugee.tiktok.common.model.dto.SlimBloggerDto;
import com.gugee.tiktok.common.model.mysql.BloggerMain;
import com.gugee.tiktok.common.model.spider.blogger.BloggerResponse;
import com.xiguaji.boot.eventbus.Event;
import com.xiguaji.boot.eventbus.EventBusTemplate;
import com.xiguaji.boot.eventbus.IEventListener;
import com.xiguaji.tiktok.task.blogger.client.AwemeClient;
import com.xiguaji.tiktok.task.blogger.spider.BloggerSpiderClient;
import com.xiguaji.tiktok.task.core.BloggerScore;
import com.xiguaji.tiktok.task.core.ScheduleTemplate;
import com.xiguaji.tiktok.task.monitor.AvgStaticstics;
import com.xiguaji.tiktok.task.monitor.LogUtil;
import com.xiguaji.tiktok.task.monitor.TimeStatistics;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

@Component
@Slf4j
public class UpdateBloggerMainEventListener implements IEventListener<UpdateBloggerMainEventListener.UpdateBloggerMainEvent> {


    final String updateBloggerMain = "UpdateBloggerMain-worker";
    final String updateBloggerMainOneThousand = "UpdateBloggerMain-top1000-worker";
//        ForkJoinPool forkJoinPool = ThreadPoolUtils.forkJoinPool(1000, updateBloggerMain);
//    ForkJoinPool forkJoinPoolOneThousand = ThreadPoolUtils.forkJoinPool(200, updateBloggerMainOneThousand);
    @Autowired
    AwemeClient awemeClient;

    @Autowired
    BloggerSpiderClient bloggerSpiderClient;

    @Autowired
    BloggerMainRepository bloggerMainRepository;

    @Autowired
    BloggerMain.BloggerMainMapper bloggerMapper;

    @Autowired
    ScheduleTemplate scheduleTemplate;

    private int queueSize=30000;
    private int corePollSize=400;
    private int maxPollSize=400;
    private ThreadPoolExecutor customPoolMain;
    private ThreadPoolExecutor customPoolMainTop;
    @PostConstruct
    public void init() {
        log.info("UpdateBloggerMainEventListener  init");
         customPoolMain = scheduleTemplate.getCustomThreadPool(updateBloggerMain, corePollSize, corePollSize, 0, queueSize, TimeUnit.MINUTES);
//        customPoolMainTop = scheduleTemplate.getCustomThreadPool(updateBloggerMainOneThousand, corePollSize, maxPollSize, 1, queueSize*3, TimeUnit.MINUTES);

//        scheduleTemplate.attach(updateBloggerMain, forkJoinPool);
//        scheduleTemplate.attach(updateBloggerMainOneThousand, forkJoinPoolOneThousand);
    }

    @Lazy
    @Autowired
    EventBusTemplate eventBusTemplate;

    @Override
    @Subscribe
    @AllowConcurrentEvents
    public void handle(UpdateBloggerMainEvent event) {
        try {
            log.info("event start");
            process(event);
            log.info("event end");
        } catch (Exception e) {
            XxlJobLogger.log(e);
            log.error("error", e);
        }
    }


    private void process(UpdateBloggerMainEvent event){

        long start = System.currentTimeMillis();

        // redis日志记录
        LogUtil.setCountSize(event.slimBloggers.size());
        ThreadPoolExecutor customThreadPool=customPoolMain;
//        if (event.isOneThousand) {
//         customThreadPool = customPoolMainTop;
//        }
        XxlJobLogger.log("CustomPool start...{}", event.slimBloggers.size());
        Collection<Future<?>> futures = new LinkedList<Future<?>>();

        final ThreadPoolExecutor customThreadPool2 = customThreadPool;
        event.slimBloggers.forEach(
                blogger -> {
                    try {
                        futures.add(customThreadPool2.submit(() -> run(event, blogger.getUid(), blogger.getRegion())));
                    } catch (Exception ex) {
                        XxlJobLogger.log(ex);
                        log.error("UpdateBloggerMain error. uid:" + blogger.getUid(), ex);
                        event.failCounts.incrementAndGet();
                    }
                }
        );
        for (Future<?> future : futures) {
            try {
                future.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
                XxlJobLogger.log(e);
            } catch (ExecutionException e) {
                e.printStackTrace();
                XxlJobLogger.log(e);
            }
        }

        log.info("===>>UpdateBloggerMainEventListener size:{} ,success:{} ,fails:{},[无视频处理的播主数={},但库表查询出的视频数={},无视频处理的总花费时间={}],use time={}", event.slimBloggers.size(), event.count.get(), event.failCounts, event.uidsWithNotHandleAweme, event.bloggerAwemesSizeFromDb, event.uidTimeWithAwemeNotUploads, (System.currentTimeMillis() - start) / 1000 + "s");
        XxlJobLogger.log("===>>UpdateBloggerMainEventListener size:{} ,success:{} ,fails:{},[无视频处理的播主数={},但库表查询出的视频数={},无视频处理的总花费时间={}],use time={}", event.slimBloggers.size(), event.count.get(), event.failCounts, event.uidsWithNotHandleAweme, event.bloggerAwemesSizeFromDb, event.uidTimeWithAwemeNotUploads, (System.currentTimeMillis() - start) / 1000 + "s");
    }
//    private void process(UpdateBloggerMainEvent event) throws ExecutionException, InterruptedException {
//
//        long start = System.currentTimeMillis();
//        // redis日志记录
//        LogUtil.setCountSize(event.slimBloggers.size());
//        ForkJoinPool currForkJoinPool=forkJoinPool;
//        XxlJobLogger.log("forkJoinPool start...{}",event.slimBloggers.size());
////        if(event.isOneThousand){
////            currForkJoinPool=forkJoinPoolOneThousand;
////        }
//        currForkJoinPool.submit(() ->
//                    event.slimBloggers.parallelStream().forEach(blogger -> {
//                        try {
//                            run(event, blogger.getUid(), blogger.getRegion());
//                        } catch (Exception ex) {
//                            XxlJobLogger.log(ex);
//                            log.error("UpdateBloggerMain error. uid:" + blogger.getUid(), ex);
//                            event.failCounts.incrementAndGet();
//                        }
//                    })
//            ).get();
//        log.info("===>>UpdateBloggerMainEventListener size:{} ,success:{} ,fails:{},[无视频处理的播主数={},但库表查询出的视频数={},无视频处理的总花费时间={}],use time={}", event.slimBloggers.size(), event.count.get(), event.failCounts, event.uidsWithNotHandleAweme, event.bloggerAwemesSizeFromDb, event.uidTimeWithAwemeNotUploads, (System.currentTimeMillis() - start) / 1000 + "s");
//        XxlJobLogger.log("===>>UpdateBloggerMainEventListener size:{} ,success:{} ,fails:{},[无视频处理的播主数={},但库表查询出的视频数={},无视频处理的总花费时间={}],use time={}", event.slimBloggers.size(), event.count.get(), event.failCounts, event.uidsWithNotHandleAweme, event.bloggerAwemesSizeFromDb, event.uidTimeWithAwemeNotUploads, (System.currentTimeMillis() - start) / 1000 + "s");
//}
//    private void process(UpdateBloggerMainEvent event) {
//
//        long start = System.currentTimeMillis();
//        // redis日志记录
//        LogUtil.setCountSize(event.slimBloggers.size());
//        XxlJobLogger.log("forkJoinPool start...{}",event.slimBloggers.size());
//                event.slimBloggers.parallelStream().forEach(blogger -> {
//                    try {
//                        run(event, blogger.getUid(), blogger.getRegion());
//                    } catch (Exception ex) {
//                        XxlJobLogger.log(ex);
//                        log.error("UpdateBloggerMain error. uid:" + blogger.getUid(), ex);
//                        event.failCounts.incrementAndGet();
//                    }
//                });
//        log.info("===>>UpdateBloggerMainEventListener size:{} ,success:{} ,fails:{},[无视频处理的播主数={},但库表查询出的视频数={},无视频处理的总花费时间={}],use time={}", event.slimBloggers.size(), event.count.get(), event.failCounts, event.uidsWithNotHandleAweme, event.bloggerAwemesSizeFromDb, event.uidTimeWithAwemeNotUploads, (System.currentTimeMillis() - start) / 1000 + "s");
//        XxlJobLogger.log("===>>UpdateBloggerMainEventListener size:{} ,success:{} ,fails:{},[无视频处理的播主数={},但库表查询出的视频数={},无视频处理的总花费时间={}],use time={}", event.slimBloggers.size(), event.count.get(), event.failCounts, event.uidsWithNotHandleAweme, event.bloggerAwemesSizeFromDb, event.uidTimeWithAwemeNotUploads, (System.currentTimeMillis() - start) / 1000 + "s");
//    }
    private void run(UpdateBloggerMainEvent event, long uid, String region) {
        TimeStatistics timeCount = new TimeStatistics();
        long start = System.currentTimeMillis();
        long updateBloggerMainEvent;
        //根据uid,region调用接口获取播主相关的数据
        BloggerResponse response = bloggerSpiderClient.get(uid, region);
        BloggerMain blogger = bloggerMapper.mapper(response.getUser());


        long bloggerSpider = System.currentTimeMillis() - start;
        timeCount.setBloggerSpider(bloggerSpider / 1000);
//        if ((System.currentTimeMillis() - start) / 1000 > 10) {
//            XxlJobLogger.log("UpdateBloggerMainEventListener blogger: " + uid + " bloggerSpiderClient.get use time:" + (System.currentTimeMillis() - start) / 1000 + "s");
//            log.info("UpdateBloggerMainEventListener blogger: " + uid + " bloggerSpiderClient.get use time:" + (System.currentTimeMillis() - start) / 1000 + "s");
//        }

        if (response.getUser() == null) {
            event.count.incrementAndGet();
            XxlJobLogger.log("***** error UpdateBloggerMain bloggerSpiderClient get empty.uid:" + uid + " region:" + region);
            log.error("UpdateBloggerMain bloggerSpiderClient get empty.uid:" + uid + " region:" + region);
            return;
        }

        long refreshBloggerAweme = System.currentTimeMillis();
        TimeStatistics tmp = awemeClient.refreshBloggerAweme(uid);
        timeCount.setAwemeSpiderBloggerAwemes(tmp.getAwemeSpiderBloggerAwemes());
        timeCount.setRefreshBloggerAweme((System.currentTimeMillis() - refreshBloggerAweme) / 1000);
        timeCount.setAwemeSize(tmp.getAwemeSize());
        timeCount.setSaveAwemeTimeStatistics(tmp.getSaveAwemeTimeStatistics());

        long getBloggerAwemeExtend = System.currentTimeMillis();
        var bloggerAwemeExtend = awemeClient.getBloggerAwemeExtend(uid);
        blogger.setOAvgLikes(bloggerAwemeExtend.getAvgLikes());
        blogger.setOAvgReviews(bloggerAwemeExtend.getAvgReviews());
        blogger.setOAvgShare(bloggerAwemeExtend.getAvgShare());
        blogger.setOAvgViews(bloggerAwemeExtend.getAvgViews());
        timeCount.setGetBloggerAwemeExtend((System.currentTimeMillis() - getBloggerAwemeExtend) / 1000);

        long getBloggerAwemes = System.currentTimeMillis();
        //todo 查询播主当前已保存的最近90d视频，与下面更新分数处理重复查询
        val bloggerAwemes = awemeClient.getBloggerAwemes(blogger.getOUid());
        timeCount.setGetBloggerAwemes((System.currentTimeMillis() - getBloggerAwemes) / 1000);

        long refreshBloggerAwemeScore = System.currentTimeMillis();
        AwemeScore awemeScore = new AwemeScore(blogger, bloggerAwemes);
        //todo 再次查询播主当前已保存的最近90d视频,并更新Score
        awemeClient.refreshBloggerAwemeScore(awemeScore);
        timeCount.setRefreshBloggerAwemeScore((System.currentTimeMillis() - refreshBloggerAwemeScore) / 1000);

        long refreshBloggerAwemeForES = System.currentTimeMillis();
        awemeClient.refreshBloggerAwemeForES(blogger.getOUid());
//        if(event.isOneThousand) {
//            event.esFutures.add(customPoolESTop.submit(() -> {
//                awemeClient.refreshBloggerAwemeForES(blogger.getOUid());
//            }));
//        }else {
//            event.esFutures.add(customPoolES.submit(() -> {
//                awemeClient.refreshBloggerAwemeForES(blogger.getOUid());
//            }));
//        }



        timeCount.setRefreshBloggerAwemeForES((System.currentTimeMillis() - refreshBloggerAwemeForES) / 1000);

        blogger.setLastPubTime(new Date());
        // 计算博主得分
        BloggerScore.ScoreParms scoreParms = new BloggerScore.ScoreParms();
        scoreParms.fans = blogger.getOFollowerCount();
        scoreParms.artnumbers = bloggerAwemes.size();
        scoreParms.comment = (double) blogger.getOAvgReviews();
        scoreParms.share = (double) blogger.getOAvgShare();
        scoreParms.like = (double) blogger.getOAvgLikes();
        blogger.setScore(BloggerScore.GetTotalScore(scoreParms));
        long updateBloggerMainTab = System.currentTimeMillis();

        var optionalBlogger = bloggerMainRepository.findByoUid(blogger.getOUid());
        var origin = optionalBlogger.get();
        blogger.setId(origin.getId());
        blogger.setCreateTime(origin.getCreateTime());
        blogger.setCreateBy(origin.getCreateBy());
        blogger.setUpdateBy("MonitorBloggerMainRecord");
        blogger.setUpdateTime(System.currentTimeMillis()/1000);
        blogger.setBloggerType(origin.getBloggerType());
        blogger.setBloggerTypeAuditStatus(origin.getBloggerTypeAuditStatus());
        blogger.setTags(origin.getTags());
        blogger.setAiAge(origin.getAiAge());
        blogger.setAiGender(origin.getAiGender());
        bloggerMainRepository.save(blogger);
        timeCount.setUpdateBloggerMainTab((System.currentTimeMillis() - updateBloggerMainTab) / 1000);

        // 记录每天播主的信息
        if(event.isOneThousand){
            var saveBloggerRecordEvent = new SaveBloggerRecordEventListener.SaveBloggerRecordEvent(event, Arrays.asList(blogger),false);
//            try {
//                Thread t=  new Thread(()->{
                    eventBusTemplate.post(saveBloggerRecordEvent);
//                }
//                );
//                t.start();
//                t.join();
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
        }

        long saveBloggerMain2ES = System.currentTimeMillis();
        val esSaveBloggerMainEvent = new ESSaveBloggerMainEventListener.ESSaveBloggerMainEvent(event, Arrays.asList(blogger));

//        try {
//            Thread t=  new Thread(()->{
                eventBusTemplate.post(esSaveBloggerMainEvent);
//             }
//            );
//            t.start();
//            t.join();
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        }

//        customPoolES.submit(()->{eventBusTemplate.post(esSaveBloggerMainEvent);});



        timeCount.setSaveBloggerMain2ES((System.currentTimeMillis() - saveBloggerMain2ES) / 1000);
//        if((System.currentTimeMillis()-start)/1000>10)
//            log.info("UpdateBloggerMainEventListener blogger: "+blogger.getOUid()+" refresh use time:"+(System.currentTimeMillis()-start)/1000+"s");

        int cnt = event.count.incrementAndGet();
        //统计当前无视频需处理的额外花费
        if (tmp.getAwemeSize() == 0) {
            event.uidsWithNotHandleAweme.incrementAndGet();
            event.uidTimeWithAwemeNotUploads.addAndGet((System.currentTimeMillis() - getBloggerAwemeExtend) / 1000);
            event.bloggerAwemesSizeFromDb.addAndGet(bloggerAwemes.size());
            if(bloggerAwemes.size()>0){
                log.warn("xxxxxxxxxxxxxxxx-----------UpdateBloggerMainEvent:接口请求视频过滤为0.但库表有视频数={},uid={}",bloggerAwemes.size(),uid);
            }

        }
        // redis日志记录
//        LogUtil.setCurSize(cnt);
        long tcnt=(System.currentTimeMillis() - start) / 1000;
        timeCount.setUpdateBloggerMainEvent(tcnt);

        if(tcnt>50) {
            log.info("******UpdateBloggerMainEvent:{}/{}, uid={},region={},awemeSize={},{}", cnt, event.slimBloggers.size(), uid, region, tmp.getAwemeSize(), timeCount);
//            XxlJobLogger.log("******UpdateBloggerMainEvent:{}/{}, uid={},region={},awemeSize={},{}", cnt, event.slimBloggers.size(), uid, region, tmp.getAwemeSize(), timeCount);
        }
        if(tcnt>50) {
            XxlJobLogger.log("******UpdateBloggerMainEvent:{}/{}, uid={},region={},awemeSize={},{}", cnt, event.slimBloggers.size(), uid, region, tmp.getAwemeSize(), timeCount);
        }
    }

    public static class UpdateBloggerMainEvent extends Event {

        public UpdateBloggerMainEvent(Event event, List<SlimBloggerDto> slimBloggers) {
            super(event, "UpdateBloggerMain");
            this.slimBloggers = slimBloggers;
        }

        public UpdateBloggerMainEvent(Event event, List<SlimBloggerDto> slimBloggers, boolean isOneThousand) {
            super(event, "UpdateBloggerMain");
            this.slimBloggers = slimBloggers;
            this.isOneThousand = isOneThousand;
        }
//        private Collection<Future<?>> esFutures = new LinkedList<Future<?>>();
        private AtomicInteger count = new AtomicInteger(0);
        private Boolean isOneThousand;
        //没有视频需处理的时间
        private AtomicLong uidTimeWithAwemeNotUploads = new AtomicLong(0);
        private AtomicLong bloggerAwemesSizeFromDb = new AtomicLong(0);
        //没有视频需处理的播主数
        private AtomicInteger uidsWithNotHandleAweme = new AtomicInteger(0);
        private AtomicInteger failCounts = new AtomicInteger(0);

        private AvgStaticstics avgStaticstics = new AvgStaticstics();

        private List<SlimBloggerDto> slimBloggers;
    }
}
